package com.pw.study.flink.sink;

import com.pw.study.flink.entities.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;


import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.ArrayList;

public class CustomSink {
    private static Object WaterSensor;

    public static void main(String[] args) {
        dealMysql();
    }

    private static void dealMysql() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
        waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
        waterSensors.add(new WaterSensor("sensor_1", 1607527996000L, 50));
        waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
        waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));

        env.setParallelism(2);
        env.fromCollection(waterSensors).addSink(new RichSinkFunction<WaterSensor>(){
            private PreparedStatement ps;
            private Connection conn;


            @Override
            public void open(Configuration parameters) throws Exception {
                // 建立到mysql的连接
                Class.forName("com.mysql.jdbc.Driver");
                conn = DriverManager.getConnection("jdbc:mysql://hadoop162:3306/test?useSSL=false", "root", "aaaaaa"); //alt+ctrl+enter
                //conn.setAutoCommit(false);
            }

            @Override
            public void close() throws Exception {
                if(conn !=null && !conn.isClosed()){
                    conn.close();
                }
            }

            @Override
            public void invoke(com.pw.study.flink.entities.WaterSensor value, Context context) throws Exception {
                String sql = "replace into sensor(id, ts, vc)values(?,?,?)";
                PreparedStatement ps = conn.prepareStatement(sql);
                // sql语句中u占位符, 需要给占位符赋值
                ps.setString(1, value.getId());
                ps.setLong(2, value.getTs());
                ps.setInt(3, value.getVc());
                // 执行写入
                ps.execute();
                ps.close();
            }
        });


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
